package com.ds.infrastructure.hyperspace.container.flow.context;

import com.ds.infrastructure.hyperspace.common.*;
import com.ds.infrastructure.hyperspace.common.config.TargetEndpointConfig;
import com.ds.infrastructure.hyperspace.common.flow.DuplexFlowContext;
import com.ds.infrastructure.hyperspace.container.endpoint.connector.Timeoutable;
import com.ds.infrastructure.hyperspace.container.endpoint.entry.AbstractEntryProtocolAdapter;
import com.ds.infrastructure.hyperspace.container.flow.HyperspaceDuplexFlowPipeline;
import com.ds.infrastructure.hyperspace.container.flow.operation.EntryFlowOperation;
import com.ds.infrastructure.hyperspace.container.internal.HyperspaceIdGenerator;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import com.ds.infrastructure.hyperspace.container.internal.opentracing.impl.HyperspaceTracerContext;
import com.ds.infrastructure.zonesdk.Zone;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateFormatUtils;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;

import static com.ds.infrastructure.hyperspace.common.StateEnum.REQUEST_TIMEOUT;
import static com.ds.infrastructure.hyperspace.common.config.TargetNodeConfig.DEFAULT_ZONE;
import static com.ds.infrastructure.hyperspace.container.internal.util.StringConstant.EMPTY_STR;
import static com.ds.infrastructure.hyperspace.container.internal.util.StringConstant.HYPHEN;

/**
 * @author  Weishaoying
 */
@Slf4j
@Getter
public class HyperspaceDuplexFlowContext implements DuplexFlowContext, Timeoutable {

    private static final String DATE_TIME_FORMATTER = "yyyy-MM-dd HH:mm:ss.SSS";

    private static final int DEFAULT_REFER_STREAM_SIZE_CAPACITY = 4;

    public static final int UNKNOWN_STATUS = -1;

    private final String requestTimeStr;

    private long requestStartTime;

    private long gatewayStartTime;

    private long gatewayCost;

    private long connectionStartTime;

    private long connectionCost;

    private long pendingStartTime;

    private long pendingCost;

    private final String requestId;

    private final NettyContainer nettyContainer;

    private final Object requestStartState;

    private final String clientIp;

    private final String forwardedIp;

    private final String realIp;

    private final EntryFlowOperation<HyperspaceDuplexFlowContext> entryFlowOperation;

    private final HyperspaceDuplexFlowContext connectionDuplexFlowContext;

    private final Map<String, TargetEndpointStream> referTargetEndpointStreamMap;

    private int payloadSize;

    private LinkedList<Object> referDataLinkList;

    @Setter
    private HyperspaceTracerContext tracerContext;

    @Setter
    private AbstractEntryProtocolAdapter entryProtocolAdapter;

    @Setter
    private EndpointProtocolEnum endpointProtocol;

    @Setter
    private String host;

    @Setter
    private String uri;

    @Setter
    private String requestType;

    @Setter
    private boolean routeMatch;

    @Setter
    private HyperspaceDuplexFlowPipeline flowHandlerPipeline;

    private StateEnum interruptState;

    @Setter
    private String interruptMessage = EMPTY_STR;

    @Setter
    private String shardingKeyType = EMPTY_STR;

    private String shardingKey = EMPTY_STR;

    @Setter
    private String zoneId = DEFAULT_ZONE;

    @Setter
    private TargetEndpoint<TargetEndpointConfig> targetEndpoint;

    private Map<String, String> contextHeader;

    @Setter
    private Object entryPrefaceState;

    private Object targetResponseState;

    private int targetResponseCode = UNKNOWN_STATUS;

    private int entryResponseCode = UNKNOWN_STATUS;

    @Setter
    private boolean inputComplete;

    @Setter
    private boolean outputComplete;

    private boolean closed;

    @Setter
    private boolean pending;

    @Setter
    private String targetRemoteAddress=HYPHEN;

    private TargetEndpointStream targetEndpointStream;

    @Setter
    private String grpcServiceRouterName;

    public HyperspaceDuplexFlowContext(NettyContainer nettyContainer,
                                       AbstractEntryProtocolAdapter entryProtocolAdapter, Object requestStartState,
                                       String forwardedIp, String realIp, String clientIp,
                                       EndpointProtocolEnum endpointProtocol,
                                       HyperspaceDuplexFlowContext connectionDuplexFlowContext) {
        this.requestStartTime = System.currentTimeMillis();
        this.requestTimeStr = DateFormatUtils.format(requestStartTime, DATE_TIME_FORMATTER);
        this.gatewayStartTime = requestStartTime;
        this.requestId = HyperspaceIdGenerator.generate();
        this.nettyContainer = nettyContainer;
        this.endpointProtocol = endpointProtocol;
        this.connectionDuplexFlowContext = connectionDuplexFlowContext;
        this.entryFlowOperation = nettyContainer.getFlowPipelineOrchestrator().getEntryFlowOperation();
        this.entryProtocolAdapter = entryProtocolAdapter;
        this.requestStartState = requestStartState;
        this.forwardedIp = forwardedIp == null ? EMPTY_STR : forwardedIp;
        this.realIp = realIp == null ? EMPTY_STR : realIp;
        this.clientIp = clientIp;
        if (connectionDuplexFlowContext != null) {
            this.referTargetEndpointStreamMap = connectionDuplexFlowContext.getReferTargetEndpointStreamMap();
        } else {
            this.referTargetEndpointStreamMap = new HashMap<>(DEFAULT_REFER_STREAM_SIZE_CAPACITY);
        }
    }

    @Override
    public void resetState() {
        this.gatewayCost = 0;
        this.payloadSize = 0;
        this.connectionCost = 0;
        this.pendingCost = 0;
    }

    @Override
    public String getServiceName() {
        return flowHandlerPipeline != null ? flowHandlerPipeline.getTargetServiceName() : "pipeline.not.found";
    }

    @Override
    public void setInterruptState(StateEnum state, String message) {
        this.interruptState = state;
        this.interruptMessage = message;
    }

    @Override
    public TargetEndpointStream getCurrentReferTargetEndpointStream() {
        if (targetEndpointStream == null) {
            targetEndpointStream = referTargetEndpointStreamMap.get(zoneId);
        }
        if (targetEndpointStream == null) {
            return null;
        }
        if (targetEndpointStream.getStatus() == TargetEndpointStreamStatus.CONNECT_FAILED) {
            referTargetEndpointStreamMap.remove(zoneId);
            return null;
        }
        return targetEndpointStream;
    }

    @Override
    public void setReferTargetEndpointStream(TargetEndpointStream stream) {
        referTargetEndpointStreamMap.putIfAbsent(zoneId, stream);
    }

    @Override
    public String getContextHeader(String headerName) {
        if (contextHeader == null) {
            return null;
        }
        return contextHeader.get(headerName);
    }

    @Override
    public void setContextHeader(String headerName, String headerValue) {
        if (contextHeader == null) {
            contextHeader = new HashMap<>(8);
        }
        contextHeader.put(headerName, headerValue);
    }

    @Override
    public void appendReferData(Object data) {
        if (referDataLinkList == null) {
            referDataLinkList = new LinkedList<>();
        }
        referDataLinkList.offer(data);
    }

    @Override
    public void close() {
        this.closed = true;
    }

    public void setShardingInfo(String shardingKeyType, String shardingKey, Zone zone) {
        this.shardingKeyType = shardingKeyType;
        this.shardingKey = shardingKey;
        this.zoneId = zone.getZoneId().toLowerCase();
    }

    public void setTargetResponseState(Object targetResponseState, int targetResponseCode) {
        this.targetResponseState = targetResponseState;
        this.targetResponseCode = targetResponseCode;
    }

    public void setEntryResponseState(int entryResponseCode) {
        this.entryResponseCode = entryResponseCode;
    }

    @Override
    public String getEventId() {
        return requestId;
    }

    @Override
    public boolean checkTimeout() {
        if (endpointProtocol.isLongLive()) {
            return false;
        }
        return System.currentTimeMillis() - requestStartTime >
                targetEndpoint.getEndpointConfig().getMaxRequestTimeout();
    }

    @Override
    public void onTimeout() {
        entryFlowOperation.interruptAndCloseContext(this, REQUEST_TIMEOUT,
                "request timeout, max timeout time [" + this.getTargetEndpoint()
                        .getEndpointConfig().getMaxRequestTimeout() + "]ms");
    }

    @Override
    public void touch() {
        if (gatewayStartTime < 1) {
            return;
        }
        gatewayStartTime = System.currentTimeMillis();
    }

    public void endGatewayDuration() {
        if (gatewayStartTime < 1) {
            return;
        }
        gatewayCost += System.currentTimeMillis() - gatewayStartTime;
        gatewayStartTime = 0;
    }

    public void touchForConnection() {
        connectionStartTime = System.currentTimeMillis();
    }

    public void endConnectionDuration() {
        if (connectionStartTime < 1) {
            return;
        }
        connectionCost += System.currentTimeMillis() - connectionStartTime;
        connectionStartTime = 0;
    }

    public void touchForPending() {
        pendingStartTime = System.currentTimeMillis();
    }

    public void endPendingDuration() {
        if (pendingStartTime < 1) {
            return;
        }
        pendingCost += System.currentTimeMillis() - pendingStartTime;
        pendingStartTime = 0;
    }

    public void addPayloadSize(int size) {
        if (size < 1) {
            return;
        }
        payloadSize += size;
    }

}
